package com.gj.xxl.job.core.executor;

import com.gj.xxl.job.core.biz.AdminBiz;
import com.gj.xxl.job.core.biz.client.AdminBizClient;
import com.gj.xxl.job.core.biz.model.TriggerParam;
import com.gj.xxl.job.core.handler.MethodJobHandler;
import com.gj.xxl.job.core.handler.annotation.GjJob;
import com.gj.xxl.job.core.server.EmbedServer;
import com.gj.xxl.job.core.thread.JobThread;
import com.gj.xxl.job.core.util.IpUtil;
import com.gj.xxl.job.core.util.NetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 该类就是执行器端的入口。
 */
public class GjJobSpringExecutor implements ApplicationContextAware, SmartInitializingSingleton {

    public static final Logger logger = LoggerFactory.getLogger(GjJobSpringExecutor.class);

    //下面这些成员变量都是定义在配置文件中的，而这里的属性，会在用户自己定义的Config配置类中被赋值成功
    //服务器的地址，也就是调度中心的地址，执行器要注册到调度中心那一端
    private String adminAddresses;
    //token令牌，这个令牌要和调度中心那一端的令牌配置成一样的，否则调度中心那端校验不通过会报错
    private String accessToken;
    //这个就是执行器的名称，注册执行器到调度中心的时候，使用的就是这个名称
    private String appname;
    //执行器的地址，这个地址在配置文件中为空，意味着使用默认地址
    //地址为：ip+port
    private String address;
    //执行器的ip地址
    private String ip;
    //端口号
    private int port;

    //下面这些方法都会在用户自己定义的Config类中被调用到
    public void setAdminAddresses(String adminAddresses) {
        this.adminAddresses = adminAddresses;
    }

    public void setAccessToken(String accessToken) {
        this.accessToken = accessToken;
    }

    public void setAppname(String appname) {
        this.appname = appname;
    }

    public void setAddress(String address) {
        this.address = address;
    }

    public void setIp(String ip) {
        this.ip = ip;
    }

    public void setPort(int port) {
        this.port = port;
    }


    //spring 容器
    private ApplicationContext applicationContext;

    @Override
    public void afterSingletonsInstantiated() {

        initJobHandlerRepository();


        //初始化admin链接路径存储集合
        //如果是在集群情况下，可能会有多个调度中心，所以，执行器要把自己分别注册到这些调度中心上
        //这里的方法就是根据用户配置的调度中心的地址，把用来远程注册的客户端初始化好
        initAdminBizList(adminAddresses, accessToken);

        //启动服务器，用于接收调度中心发过来的通知
        initEmbedServer(address, ip, port, appname, accessToken);
    }


    //该成员变量是用来存放AdminBizClient对象的，而该对象是用来向调度中心发送注册信息的
    private static List<AdminBiz> adminBizList;

    private void initAdminBizList(String adminAddresses, String accessToken) {
        if (adminAddresses != null && adminAddresses.trim().length() > 0) {
            //在这里判断可能有多个调度中心服务器，所以要展开遍历
            for (String address : adminAddresses.trim().split(",")) {
                if (address != null && address.trim().length() > 0) {
                    //根据服务器地址和令牌创建一个客户端
                    AdminBiz adminBiz = new AdminBizClient(address.trim(), accessToken);
                    //如果AdminBizClient对象为空，就初始化集合对象
                    if (adminBizList == null) {
                        adminBizList = new ArrayList<AdminBiz>();
                    }
                    //把创建好的客户端添加到集合中
                    adminBizList.add(adminBiz);
                }
            }
        }
    }

    public static List<AdminBiz> getAdminBizList() {
        return adminBizList;
    }

    private void initEmbedServer(String address, String ip, int port, String appname, String accessToken) {
        //这里就是使用默认地址的操作，因为执行器一端的配置文件中并没有配置执行器的地址
        //所以这里使用工具类得到默认端口号9999
        port = port > 0 ? port : NetUtil.findAvailablePort(9999);
        //在这里得到默认IP地址
        ip = (ip != null && ip.trim().length() > 0) ? ip : IpUtil.getIp();
        //判断地址是否为null
        if (address == null || address.trim().length() == 0) {
            //如果为空说明真的没有配置，那就把刚才得到的IP地址和port拼接起来
            //得到默认的执行器地址
            String ip_port_address = IpUtil.getIpPort(ip, port);
            address = "http://{ip_port}/".replace("{ip_port}", ip_port_address);
        }
        //校验token
        if (accessToken == null || accessToken.trim().length() == 0) {
            logger.warn(">>>>>>>>>>> xxl-job accessToken is empty. To ensure system security, please set the accessToken.");
        }
        //创建执行
        new EmbedServer().start(address, port, appname, accessToken);
    }

    private static ConcurrentHashMap<String, MethodJobHandler> jobHandlerRepository = new ConcurrentHashMap<>();

    //遍历 spring 容器中的所有 bean，然后遍历 bean 中的每个方法，找出方法上加了 @GjJob 注解的方法，拿到 Method 和 bean 对象

    private void initJobHandlerRepository() {
        String[] beanDefinitionNames = applicationContext.getBeanDefinitionNames();
        for (String beanDefinitionName : beanDefinitionNames) {
            Object bean = applicationContext.getBean(beanDefinitionName);
            Method[] methods = bean.getClass().getDeclaredMethods();
            for (Method method : methods) {
                if (method.isAnnotationPresent(GjJob.class)) {
                    GjJob gjJob = method.getAnnotation(GjJob.class);
                    method.setAccessible(true);
                    jobHandlerRepository.put(gjJob.value(), new MethodJobHandler(bean, method));
                }
            }
        }
    }

    //jobId 和 JobThread 的映射关系
    private static ConcurrentHashMap<Integer, JobThread> jobThreadMap = new ConcurrentHashMap<>();
    private static AtomicInteger index = new AtomicInteger(0);

    public static JobThread loadJobThread(Integer jobId) {
        return jobThreadMap.get(jobId);
    }

    public static JobThread registerJobThread(Integer jobId) {
        JobThread jobThread = new JobThread(jobId);
        jobThread.setName("jobThread" + index.incrementAndGet());
        jobThread.setDaemon(true);
        jobThread.start();
        JobThread oldJobThread = jobThreadMap.put(jobId, jobThread);

        //执行具体的阻塞处理策略
        //这里是抛弃策略：抛弃上一次还没执行完的任务，直接执行这一次任务
        if (oldJobThread != null) {
            oldJobThread.toStop();
            oldJobThread.interrupt();
        }
        return jobThread;
    }

    public static void removeJobThread(Integer jobId) {
        JobThread removedJobThread = jobThreadMap.remove(jobId);
        if (removedJobThread != null) {
            removedJobThread.toStop();
            removedJobThread.interrupt();
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    public static MethodJobHandler getMethodJobHandler(TriggerParam triggerParam) {
        return jobHandlerRepository.get(triggerParam.getExecutorHandler());
    }
}
